package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.CartAddUuBean;
import com.atguigu.edu.realtime.utils.ClickHouseUtil;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ClassName DwsTradeCartAddUuWindow
 * @Description TODO
 * @Author$ 邢家俊
 * @Date 2023-5-8 11:26
 * @Version 1.0
 **/
public class DwsTradeCartAddUuWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //3.1 声明消费主题以及消费者组
        String topic = "dwd_trade_cart_add";
        String groupId = "dws_trade_cart_add_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> jsonObjWithWatermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );

      //  jsonObjWithWatermarkDS.print("=====");

        //TODO 6.按照用户id进行分组
        KeyedStream<JSONObject, String> keyedDS
                = jsonObjWithWatermarkDS.keyBy(jsonObj -> jsonObj.getString("user_id"));

        //TODO 7.使用Flink的状态编程  过滤出独立用户
        SingleOutputStreamOperator<JSONObject> keybyuidDS = keyedDS.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<String> lastCartAddState;
            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastCartAddDateState", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                this.lastCartAddState
                        = getRuntimeContext().getState(valueStateDescriptor);
            }
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                String lastCourseName = lastCartAddState.value();

                if (StringUtils.isEmpty(lastCourseName) || !lastCourseName.equals(value.getString("course_name"))) {
                    out.collect(value);
                    lastCartAddState.update(lastCourseName);
                }
            }
        });
        //keybyuidDS.print("---------");
        AllWindowedStream<JSONObject, TimeWindow> windowallDS
                = keybyuidDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //{"course_id":93,"cart_price":200.00,"create_time":"2023-05-08 11:41:08","user_id":8,"course_name":"CSS3特效实战",
        // "id":12241,"ts":"1683517267"}

        SingleOutputStreamOperator<CartAddUuBean> aggregateDS = windowallDS.aggregate(new AggregateFunction<JSONObject, Long, Long>() {
            @Override
            public Long createAccumulator() {
                return 0L;
            }
            @Override
            public Long add(JSONObject value, Long accumulator) {
                return ++accumulator;
            }
            @Override
            public Long getResult(Long accumulator) {
                return accumulator;
            }
            @Override
            public Long merge(Long a, Long b) {
                return null;
            }
        }, new AllWindowFunction<Long, CartAddUuBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<Long> values, Collector<CartAddUuBean> out) throws Exception {
                String stt = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                for (Long value : values) {
                    out.collect(new CartAddUuBean(stt, edt, value, System.currentTimeMillis()));
                }
            }
        });
        aggregateDS.addSink(ClickHouseUtil.getSinkFunction("insert into  dws_trade_cart_add_uu_window values(?,?,?,?)"));
        env.execute();
    }
}
